package com.gzyj.flink.gps.map.noposition;

import cn.hutool.core.lang.Pair;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.bestvike.linq.Linq;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.dto.alarm.Kafka_VehicleRemindContent;
import com.gzyj.flink.dto.alarm.NoPositionData;
import com.gzyj.flink.dto.alarm.RemindContent;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.map.fatiguedriving.FatigueDrivingResult;
import com.gzyj.flink.gps.mapper.mysql.MonWushujuSetMapper;
import com.gzyj.flink.gps.mapper.mysql.TsAlarmRuleMapper;
import com.gzyj.flink.gps.service.NormalAlarmService;
import com.gzyj.flink.gps.utils.TimeUtils;
import com.gzyj.flink.pojo.mysql.set.MonWushujuSet;
import com.gzyj.flink.pojo.mysql.set.TsAlarmRule;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.Date;
import java.util.concurrent.TimeUnit;

public class NoPositionDataProcess extends RichFlatMapFunction<GPSPosition, FatigueDrivingResult> {
    LettuceConnectionFactory redisFactory;

    SqlSessionFactory mysql;

    NormalAlarmService normalAlarmService;

    private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisSessionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        redisFactory = RedisSessionFactory.getRedisFactory();
        mysql = MySQLMybatisSessionFactory.getSqlSessionFactory();
        normalAlarmService = new NormalAlarmService();
    }


    private NoPositionData TryGetData_NPD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {

        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_NoPosition + vehId);

        if (obj == null) {
            return new NoPositionData();
        } else {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), NoPositionData.class);
        }


    }

    public Pair<Boolean, MonWushujuSet> TryGetRuleSetting(GPSPosition vehicle, MonWushujuSetMapper monWushujuSetMapper, TsAlarmRuleMapper tsAlarmRuleMapper) {
        MonWushujuSet setting = getVehicleForNoPositionParamsDic(vehicle.getVehId(), monWushujuSetMapper);
        if (setting != null) {
            return new Pair<>(true, setting);
        } else {
            TsAlarmRule rule = tsAlarmRuleMapper.selectByRuleTypeWushuju(vehicle.getShiyongxingzhi());
            if (rule == null || rule.getRuleFlag() == 0) {
                return new Pair<>(false, null);
            } else {
                setting = getVehicleForNoPositionParamsDic(rule.getRuleNo(), monWushujuSetMapper);
                if (setting != null) {
                    return new Pair<>(true, setting);
                } else {
                    return new Pair<>(false, null);
                }
            }


        }
    }

    private MonWushujuSet getVehicleForNoPositionParamsDic(String vehId, MonWushujuSetMapper monWushujuSetMapper) {

        return monWushujuSetMapper.selectByVehID(vehId);

    }

    public Driverlogininfo TryGetData_DLD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {
        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_DriverSignOn + vehId);
        if (obj == null) {
            return new Driverlogininfo();
        } else {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Driverlogininfo.class);
        }

    }

    private RemindContent GenerateMessage(NoPositionData data, MonWushujuSet setting, GPSPosition gps, String msgType) {

        if (msgType.equalsIgnoreCase("预警短信")) {
            RemindContent content = new RemindContent() {
                {
                    VehicleId = gps.getVehId();
                    Ipaddress = gps.getTerminalPhoneNo();
                    VehicleNo = gps.getCheliangpaizhao();
                    PlateColor = gps.getChepaiyanse();
                    RemindNote = setting.getTxnr();
                    RemindType = 0x01;
                    RemindTime = new Date();
                    MessageType = "无数据报警";
                    AlarmNote = "无数据报警";
                    AlarmGuid = data.SingleGuid;
                    GpsTime = data.GpsTime;
                    LastSpeed = gps.getSpeedExt();
                }

            };
            return content;
        } else if (msgType.equalsIgnoreCase("报警短信")) {
            RemindContent content = new RemindContent() {
                {
                    VehicleId = gps.getVehId();
                    Ipaddress = gps.getTerminalPhoneNo();
                    VehicleNo = gps.getCheliangpaizhao();
                    PlateColor = gps.getChepaiyanse();
                    RemindNote = setting.getTxnr();
                    RemindType = 0x02;
                    RemindTime = new Date();
                    MessageType = "无数据报警";
                    AlarmNote = "无数据报警";
                    AlarmGuid = data.SingleGuid;
                    GpsTime = data.GpsTime;
                    LastSpeed = gps.getSpeedExt();
                }

            };
            return content;
        } else {
            return null;
        }

    }

    private Kafka_VehicleRemindContent DisposeMessage(NoPositionData noPositionData, MonWushujuSet mySetting, GPSPosition gps, String msgType) {
        if (msgType.contentEquals("预警短信")) {
        } else if (msgType.equalsIgnoreCase("报警短信")) {
            if (mySetting.getTxcs() > 0 && StringUtils.isNotEmpty(mySetting.getTxnr())) {
                long ts = TimeUtils.getMinusSecond(new Date(), noPositionData.AlarmContentSendTime);
                if (ts >= mySetting.getTxjg() && noPositionData.AlarmContentCount < mySetting.getTxcs()) {
                    noPositionData.AlarmContentCount++;
                    noPositionData.AlarmContentSendTime = new Date();
                    RemindContent message = GenerateMessage(noPositionData, mySetting, gps, msgType);
                    return normalAlarmService.SendRemindToKafka(message);
                }
            }
        }

        return null;
    }


    @Override
    public void flatMap(GPSPosition gpsPosition, Collector<FatigueDrivingResult> collector) throws Exception {


        RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);
        SqlSession sqlSession = mysql.openSession(true);

        try {

            MonWushujuSetMapper monWushujuSetMapper = sqlSession.getMapper(MonWushujuSetMapper.class);
            TsAlarmRuleMapper tsAlarmRuleMapper = sqlSession.getMapper(TsAlarmRuleMapper.class);

            Driverlogininfo driverLoginData = TryGetData_DLD(gpsPosition.getVehId(), clusterRedisTemplate);
            NoPositionData noPositionData = TryGetData_NPD(gpsPosition.getVehId(), clusterRedisTemplate);

            if (noPositionData == null) return;

            Pair<Boolean, Integer> p = TimeUtils.CheckData(noPositionData.getGpsTime(), gpsPosition.getTIME());
            int i = p.getValue();
            if (!p.getKey()) return;

            //获取运算参数
            Pair<Boolean, MonWushujuSet> p1 = TryGetRuleSetting(gpsPosition, monWushujuSetMapper, tsAlarmRuleMapper);
            MonWushujuSet mySetting = p1.getValue();
            if (p1.getKey()) {
                //如果车辆配置了规则
                if (mySetting.getEnable() == 1) {
                    if (!noPositionData.IsCalc) {
                        noPositionData.ResetData();
                        noPositionData.StartCalc(gpsPosition);
                    }
                    noPositionData.WriteCalc(gpsPosition);
                    double timeSpan = noPositionData.CalcConditionTimeSpan();
                    double distance = noPositionData.CalcConditionDistance();
                    if (timeSpan > mySetting.getZxcxsj() * 60 && distance > mySetting.getZxydjl()) {
                        // 两次位置上报时间超过最小上报时间且两次位置距离超过最小距离
                        if (!noPositionData.IsAlarm) {
                            noPositionData.StartAlarm(gpsPosition);
                            //从没报警切换到报警，发送开始报警
                            collector.collect(new FatigueDrivingResult(normalAlarmService.SendAlarmToKafka(noPositionData, gpsPosition, driverLoginData, true), null));
                        }
                        //其他的时候记录报警的过程数据
                        noPositionData.WriteAlarm(gpsPosition);
                        //停止报警
                        if (noPositionData.IsAlarm) {
                            noPositionData.StopAlarm(gpsPosition);
                            //从报警切换到没报警，发送停止报警
                            collector.collect(new FatigueDrivingResult(normalAlarmService.SendAlarmToKafka(noPositionData, gpsPosition, driverLoginData, false), DisposeMessage(noPositionData, mySetting, gpsPosition, "报警短信")));
                        }
                        noPositionData.GenerateSingleGuid();
                        noPositionData.ResetData();
                    } else {
                        noPositionData.ResetData();
                        noPositionData.StartCalc(gpsPosition);
                        noPositionData.WriteCalc(gpsPosition);
                    }
                } else {
                    //如果车辆规则关闭
                    if (noPositionData.IsAlarm) {
                        noPositionData.StopAlarm(gpsPosition);
                        //从报警切换到没报警，发送停止报警
                        collector.collect(new FatigueDrivingResult(normalAlarmService.SendAlarmToKafka(noPositionData, gpsPosition, driverLoginData, false), null));
                    }
                    noPositionData.GenerateSingleGuid();
                    noPositionData.ResetData();
                }
            } else {
                //如果车辆没有配置规则或者取消了规则
                if (noPositionData.IsAlarm) {
                    noPositionData.StopAlarm(gpsPosition);
                    //从报警切换到没报警，发送停止报警
                    collector.collect(new FatigueDrivingResult(normalAlarmService.SendAlarmToKafka(noPositionData, gpsPosition, driverLoginData, false), null));
                }
                noPositionData.GenerateSingleGuid();
                noPositionData.ResetData();
            }

            clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_NoPosition + gpsPosition.getVehId(), noPositionData, 8, TimeUnit.HOURS);

        } catch (Exception e) {
            e.printStackTrace();
            sqlSession.rollback();
        } finally {
            sqlSession.close();
        }

    }
}
